package MsgBus

import (
    "sync"

    "github.com/vmihailenco/msgpack"
    "go.uber.org/zap"
)

type MessageBus struct {
    store  map[string]*container
    cnt    int64
    mu     sync.RWMutex
    logger *zap.Logger
}

func New() *MessageBus {
    x := &MessageBus{}
    x.store = make(map[string]*container)
    return x
}

func (s *MessageBus) WithLogger(l *zap.Logger) *MessageBus {
    s.logger = l
    return s
}

func (s *MessageBus) Sub(topics ...string) *Subscriber {
    s.mu.Lock()
    defer s.mu.Unlock()

    suber := newSubscriber(64, s.cnt, topics)
    s.cnt += 1

    for _, topic := range topics {
        cnt, ok := s.store[topic]
        if ok {

            cnt.add(suber)
        } else {

            new_cnt := newContianer()
            s.store[topic] = new_cnt
            new_cnt.add(suber)
        }
    }
    return suber
}

func (s *MessageBus) UnSub(pipes ...*Subscriber) {
    s.mu.Lock()
    defer s.mu.Unlock()

    for _, pipe := range pipes {
        for _, topic := range pipe.topics {

            container, ok := s.store[topic]
            if !ok {
                continue
            }
            if container.remove(pipe) == 0 {
                delete(s.store, topic)
            }
        }
        pipe.markDeleted()
    }
}

func (s *MessageBus) Pub(topic string, msg []byte) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    cnt, ok := s.store[topic]
    if !ok {
        return
    }

    cnt.iterate(func(pipe *Subscriber) {
        _ = pipe.Push(msg)
    })
}

func (s *MessageBus) PubObject(topic string, obj interface{}) {
    b, err := msgpack.Marshal(obj)
    if err != nil {
        return
    }
    s.Pub(topic, b)
}
